/**
 * Software License, Version 1.0 Copyright 2003 The Trustees of Indiana
 * University. All rights reserved. Redistribution and use in source and binary
 * forms, with or without modification, are permitted provided that the
 * following conditions are met: 1) All redistributions of source code must
 * retain the above copyright notice, the list of authors in the original source
 * code, this list of conditions and the disclaimer listed in this license; 2)
 * All redistributions in binary form must reproduce the above copyright notice,
 * this list of conditions and the disclaimer listed in this license in the
 * documentation and/or other materials provided with the distribution; 3) Any
 * documentation included with all redistributions must include the following
 * acknowledgement: "This product includes software developed by the Community
 * Grids Lab. For further information contact the Community Grids Lab at
 * http://communitygrids.iu.edu/." Alternatively, this acknowledgement may
 * appear in the software itself, and wherever such third-party acknowledgments
 * normally appear. 4) The name Indiana University or Community Grids Lab or
 * NaradaBrokering, shall not be used to endorse or promote products derived
 * from this software without prior written permission from Indiana University.
 * For written permission, please contact the Advanced Research and Technology
 * Institute ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202. 5)
 * Products derived from this software may not be called NaradaBrokering, nor
 * may Indiana University or Community Grids Lab or NaradaBrokering appear in
 * their name, without prior written permission of ARTI. Indiana University
 * provides no reassurances that the source code provided does not infringe the
 * patent or any other intellectual property rights of any other entity. Indiana
 * University disclaims any liability to any recipient for claims brought by any
 * other entity based on infringement of intellectual property rights or
 * otherwise. LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH
 * NO WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY
 * GIVES NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF
 * INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS.
 * INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS",
 * "VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.
 * LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR
 * ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION
 * GENERATED USING SOFTWARE.
 */
package cgl.narada.util;

import java.util.Vector;

/**
 * A message queue. Each message is an array of bytes. By default, the queue
 * size will grow infinitely.
 * 
 * @author John Yin
 * @author Hongbin Liu
 * @author Harshawardhan Gadgil
 * @version 1.0
 */

public class DataQueue {

	private Vector m_dataArray = new Vector();
	private int m_currentSize = 0;
	private int m_size = 0;

	private static final byte[] NullByteArray = new byte[0];
	private static final DataRecord[] NullDataRecordArray = new DataRecord[0];

	private boolean aborted = false;

	/**
	 * This method is used by HTTP Transport threads. It assumes that data added
	 * consists of one or more messages, each of them led by the message length.
	 * Added by Hongbin
	 */
	public synchronized void addAll(byte[] data) {
		if (data == null || data.length == 0) return;
		int len = 0;
		int pos = 0;
		while (pos < data.length) {
			len = ByteUtilities.getInt(data, pos);
			// System.out.println("Dataqueue addAll, len " + len);
			byte[] temp = new byte[len];
			pos += 4;
			System.arraycopy(data, pos, temp, 0, len);
			addData(temp);
			pos += len;
		}
	}

	/**
	 * This method is used by HTTP Transport threads, which buffers messages
	 * until it reaches a certain size; then just send all in the buffer in one
	 * HTTP post. Note byte array it returns includes length for each message.
	 * Added by Hongbin
	 */
	public synchronized byte[] removeAll() {
		while (m_dataArray.size() == 0 && !aborted) {
			try {
				wait();
				// System.out.println("dataqueue, removeAll waiting ...");
			}
			catch (InterruptedException ie) {
				ie.printStackTrace();
				return NullByteArray;
			}
		}
		if (m_dataArray.size() == 0) return NullByteArray;

		java.io.ByteArrayOutputStream baos = new java.io.ByteArrayOutputStream(
				m_currentSize);
		int removeSize = 0;

		// HG: MUST use remove(0) to remove items from front of queue..
		// Enumeration removes items in random order

		while (true) {

			// Quick check to make sure that there are elements in the queue
			if (m_dataArray.size() == 0) break;

			DataRecord dr = (DataRecord) m_dataArray.elementAt(0);
			byte[] data = dr.m_payload;
			removeSize += (4 + data.length);
			if (removeSize >= 10000000) break;
			try {
				// System.out.println("removAll, removing " + data.length);

				baos.write(ByteUtilities.getBytes(data.length));
				baos.write(data);
			}
			catch (java.io.IOException ioe) {
				ioe.printStackTrace();
			}
			m_dataArray.remove(dr);
			m_currentSize -= data.length;
		}

		notify();
		// System.out.println("removeAll return " + baos.toByteArray().length +"
		// bytes");
		return baos.toByteArray();

	}

	public synchronized DataRecord[] removeAll2() {
		while (m_dataArray.size() == 0 && !aborted) {
			try {
				wait();
			}
			catch (InterruptedException ie) {
				ie.printStackTrace();
				return NullDataRecordArray;
			}
		}
		if (m_dataArray.size() == 0) return NullDataRecordArray;
		m_currentSize = 0;
		m_dataArray.clear();
		notify();
		return (DataRecord[]) m_dataArray.toArray();
	}

	/**
	 * ------------------------------------------------------------ Sets the
	 * maximum # of bytes the queu can handle before blocking on add. 0 is
	 * infinite.
	 */
	public void setMaxQueueSize(int size) {
		m_size = size;
	}

	/**
	 * ------------------------------------------------------------
	 */
	public int getMaxQueueSize() {
		return m_size;
	}

	/**
	 * ------------------------------------------------------------
	 */
	public int getQueueSize() {
		return m_currentSize;
	}

	/**
	 * ------------------------------------------------------------
	 */
	public int getNumMessages() {
		return m_dataArray.size();
	}

	/**
	 * ------------------------------------------------------------ Remove a
	 * piece of data from the data queue. Wait infinitely for it.
	 */
	public byte[] removeData() {
		return removeData(0);
	}

	/**
	 * ------------------------------------------------------------ Will block
	 * until new data segment is available.
	 * 
	 * @param waitTime
	 *            wait this number of milliseconds. If nothing appears then
	 *            return null.
	 */
	public synchronized byte[] removeData(long waitTime) {
		while (m_dataArray.size() == 0 && !aborted) {
			try {
				wait(waitTime);
			}
			catch (InterruptedException e) {
				return null;
			}
		}

		if (m_dataArray.size() == 0) {
			return null;
		}

		DataRecord data = (DataRecord) m_dataArray.firstElement();
		m_dataArray.removeElementAt(0);
		m_currentSize -= data.m_payload.length;
		notify();
		return data.m_payload;
	}

	/**
	 * ------------------------------------------------------------
	 */
	public synchronized void addData(byte[] data) {
		while (m_size != 0 && (m_currentSize > m_size)) {
			try {
				wait();
				System.out.println("dataqueue, addData waiting ...");
			}
			catch (InterruptedException e) {}
		}

		DataRecord dataR = new DataRecord();
		dataR.m_queueTime = System.currentTimeMillis();
		dataR.m_payload = data;
		m_currentSize += data.length;
		m_dataArray.addElement(dataR);

		notify();
	}

	/**
	 * ------------------------------------------------------------
	 */
	public synchronized void dispose() {
		aborted = true; // added on June 30, 2004
		notifyAll();
	}
}

/**
 * ===============================================================
 */
class DataRecord {
	long m_queueTime = 0;
	byte[] m_payload = null;
}
